package master

import (
	"fmt"
	"mapreduce-go/call"
	"net"
	"net/rpc"
	"os"
	"sync"
)

func (mr *Master) StartRpcServer() {
	server := rpc.NewServer()

	// 注册master方法集
	server.Register(mr)
	os.Remove(mr.Address)

	listener, err := net.Listen("tcp", mr.Address)
	if err != nil {
		fmt.Println("error : ", err.Error())
	}

	mr.Listener = listener

	go func() {
		for {
			select {
			case <-mr.Shutdown:
				return
			default:
				coon, err := mr.Listener.Accept()
				if err != nil {
					fmt.Println("error : ", err.Error())
					return
				}

				go func() {
					server.ServeConn(coon)
					coon.Close()
				}()

			}
		}

	}()
}

// 紧急中断
func (mr *Master) ShutdownServer(_, _ *struct{}) error {
	fmt.Println("shutdown rpc server")
	close(mr.Shutdown)
	mr.Listener.Close()
	return nil
}

// 正常停止master
func (mr *Master) StopServer() {
	var reply call.ShutdownReply

	ok := call.Call(mr.Address, "Master.ShutdownServer", new(struct{}), &reply)

	if !ok {
		fmt.Println("error")
	}

	fmt.Println("shutdown server done")
}

// 实现一个worker注册函数 这是一个rpc方法
func (mr *Master) Register(args *call.RegisterArgs, _ *struct{}) error {
	mr.Lock()

	defer mr.Unlock()

	// 注册worker
	mr.Workers = append(mr.Workers, args.WorkerAddr)
	mr.NewCond.Broadcast()

	fmt.Println("master : register worker")

	return nil
}

// 实现一个传递函数 将所有已存在的workers与新注册的worker传递到一个通道中
// 让调度函数进行接收处理
func (mr *Master) ForwardRegistration(ch chan *string) {
	var lock = &sync.Mutex{}

	i := 0
	for {
		lock.Lock()

		if len(mr.Workers) > i {
			i++
			fmt.Println("master : find worker : ", mr.Workers[i-1])
			ch <- &mr.Workers[i-1]
		} else {
			mr.Lock()
			mr.NewCond.Wait()
			mr.Unlock()
		}

		lock.Unlock()
	}
}

// 该函数通过向每个worker发送rpc请求来清理worker
// 返回worker中 worker已经执行的任务数量
func (mr *Master) KillWorker() []int {
	mr.Lock()
	defer mr.Unlock()

	var ntask []int

	for _, w := range mr.Workers {
		fmt.Println("shutdown worker : ", w)
		var reply call.ShutdownReply
		// 调用worker的shutdown方法

		ok := call.Call(w, "Worker.Shutdown", new(struct{}), &reply)

		if !ok {
			fmt.Println("shutdown failed")
		} else {
			ntask = append(ntask, reply.Id)
		}
	}

	return ntask
}
